package com.jzl.canal_consumer.etl.et.rocketmq.connector;

import com.alibaba.otter.canal.client.rocketmq.ConsumerBatchMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials;
import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook;
import com.jzl.util.LoggerUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @param <T> 队列消息数据实体类型
 * @author zwwang
 */
public abstract class AbstractMqCanalConnector<T> implements CanalRocketMqConnector<T>
{
    private String nameServer;
    private String[] topic;
    private String groupName;
    private volatile boolean connected;
    private DefaultMQPushConsumer rocketMQConsumer;
    private BlockingQueue<ConsumerBatchMessage<T>> messageBlockingQueue;
    private long batchProcessTimeout;
    private volatile ConsumerBatchMessage<T> lastGetBatchMessage;
    private String accessKey;
    private String secretKey;

    protected AbstractMqCanalConnector(String nameServer, String groupName, String... topic)
    {
        this.connected = false;
        this.batchProcessTimeout = 60000L;
        this.lastGetBatchMessage = null;
        this.nameServer = nameServer;
        this.topic = topic;
        this.groupName = groupName;
        this.messageBlockingQueue = new LinkedBlockingQueue<>(1024);
    }

    protected AbstractMqCanalConnector(String nameServer, String groupName, String accessKey, String secretKey, String... topic)
    {
        this(nameServer, groupName, topic);
        this.accessKey = accessKey;
        this.secretKey = secretKey;
    }

    @Override
    public void connect() throws CanalClientException
    {
        RPCHook rpcHook = null;
        if (null != this.accessKey && this.accessKey.length() > 0 && null != this.secretKey && this.secretKey.length() > 0)
        {
            SessionCredentials sessionCredentials = new SessionCredentials();
            sessionCredentials.setAccessKey(this.accessKey);
            sessionCredentials.setSecretKey(this.secretKey);
            rpcHook = new ClientRPCHook(sessionCredentials);
        }

        this.rocketMQConsumer = new DefaultMQPushConsumer(this.groupName, rpcHook, new AllocateMessageQueueAveragely());
        this.rocketMQConsumer.setVipChannelEnabled(false);
        if (!StringUtils.isBlank(this.nameServer))
        {
            this.rocketMQConsumer.setNamesrvAddr(this.nameServer);
        }
    }

    @Override
    public void disconnect() throws CanalClientException
    {
        this.rocketMQConsumer.shutdown();
        this.connected = false;
    }

    @Override
    public boolean checkValid() throws CanalClientException
    {
        return this.connected;
    }

    @Override
    public synchronized void subscribe(String filter) throws CanalClientException
    {
        if (!this.connected)
        {
            try
            {
                if (this.rocketMQConsumer == null)
                {
                    this.connect();
                }
                if (this.topic != null && this.topic.length > 0)
                {
                    for (String s : this.topic)
                    {
                        this.rocketMQConsumer.subscribe(s, filter);
                    }
                }
                this.rocketMQConsumer.registerMessageListener((MessageListenerOrderly) (messageExts, context) ->
                {
                    context.setAutoCommit(true);
                    boolean isSuccess = AbstractMqCanalConnector.this.process(messageExts);
                    return isSuccess ? ConsumeOrderlyStatus.SUCCESS : ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                });
                this.rocketMQConsumer.start();
            } catch (MQClientException var3)
            {
                this.connected = false;
                LoggerUtil.error("Start RocketMQ consumer error", var3);
            }

            this.connected = true;
        }
    }

    private boolean process(List<MessageExt> messageExts)
    {
        LoggerUtil.debug("Get Message:{}", messageExts);
        List<T> messageList = com.alibaba.google.common.collect.Lists.newArrayList();
        for (MessageExt messageExt : messageExts)
        {
            byte[] data = messageExt.getBody();
            if (data != null)
            {
                try
                {
                    T t = deserializerMsgData(data);
                    if (t != null)
                    {
                        messageList.add(t);
                    }
                } catch (Exception var9)
                {
                    LoggerUtil.error("deserializerMsgData error", var9);
                    throw new CanalClientException(var9);
                }
            } else
            {
                LoggerUtil.warn("Received message data is null");
            }
        }

        //把解析后的数据压进阻塞队列
        ConsumerBatchMessage<T> batchMessage = new ConsumerBatchMessage<>(messageList);
        try
        {
            this.messageBlockingQueue.put(batchMessage);
        } catch (InterruptedException var8)
        {
            LoggerUtil.error("Put message to queue error", var8);
            throw new RuntimeException(var8);
        }
        //阻塞队列等待消费，超时时间为60秒。
        boolean isCompleted;
        try
        {
            isCompleted = batchMessage.waitFinish(this.batchProcessTimeout);
        } catch (InterruptedException var7)
        {
            LoggerUtil.error("Interrupted when waiting messages to be finished.", var7);
            throw new RuntimeException(var7);
        }
        boolean isSuccess = batchMessage.isSuccess();
        return isCompleted && isSuccess;
    }

    @Override
    public void subscribe() throws CanalClientException
    {
        this.subscribe(null);
    }

    @Override
    public void unsubscribe() throws CanalClientException
    {
        for (String s : this.topic)
        {
            this.rocketMQConsumer.unsubscribe(s);
        }
    }

    @Override
    public List<T> getList(Long timeout, TimeUnit unit) throws CanalClientException
    {
        List<T> messages = this.getListWithoutAck(timeout, unit);
        if (messages != null && !messages.isEmpty())
        {
            this.ack();
        }
        return messages;
    }

    @Override
    public List<T> getListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException
    {
        try
        {
            if (this.lastGetBatchMessage != null)
            {
                throw new CanalClientException("mq get/ack not support concurrent & async ack");
            }
            ConsumerBatchMessage<T> batchMessage = this.messageBlockingQueue.poll(timeout, unit);
            if (batchMessage != null)
            {
                this.lastGetBatchMessage = batchMessage;
                return batchMessage.getData();
            }
        } catch (InterruptedException var4)
        {
            LoggerUtil.warn("Get message timeout", var4);
            throw new CanalClientException("Failed to fetch the data after: " + timeout);
        }
        return com.alibaba.google.common.collect.Lists.newArrayList();
    }

    @Override
    public void ack() throws CanalClientException
    {
        try
        {
            if (this.lastGetBatchMessage != null)
            {
                this.lastGetBatchMessage.ack();
            }
        } catch (Throwable var5)
        {
            if (this.lastGetBatchMessage != null)
            {
                this.lastGetBatchMessage.fail();
            }
        } finally
        {
            this.lastGetBatchMessage = null;
        }

    }

    @Override
    public void rollback() throws CanalClientException
    {
        try
        {
            if (this.lastGetBatchMessage != null)
            {
                this.lastGetBatchMessage.fail();
            }
        } finally
        {
            this.lastGetBatchMessage = null;
        }

    }

    @Override
    public String getGroupName()
    {
        return groupName;
    }

    @Override
    public String[] getTopic()
    {
        return topic;
    }

    @Override
    public String getNameServer()
    {
        return nameServer;
    }

    @Override
    public Message get(int batchSize) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    @Override
    public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    @Override
    public Message getWithoutAck(int batchSize) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    @Override
    public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    @Override
    public void ack(long batchId) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    @Override
    public void rollback(long batchId) throws CanalClientException
    {
        throw new CanalClientException("mq not support this method");
    }

    /**
     * 反序列化队列消息
     *
     * @param data 队列消息二进制数组
     * @return 反序列化实体
     */
    protected abstract T deserializerMsgData(byte[] data);
}
